package com.example.testpulsar.mq.factory;

import com.example.testpulsar.mq.configforpulsar.PulsarConfiguration;
import com.example.testpulsar.mq.configforpulsar.PulsarProperties;
import com.example.testpulsar.mq.constant.MqConstant;
import com.example.testpulsar.mq.sender.MqMessageSender;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;

@Configuration
@ConditionalOnBean(PulsarClient.class)
public class MessageSenderFactory {

    @Resource
    PulsarClient pulsarClient;

    @Resource
    PulsarProperties pulsarProperties;

    @Bean(name = "wxgzhHistoryListAcqSender")
    public MqMessageSender getWXGZHHistoryListAcqSender() throws PulsarClientException {
        Producer<String> producer = PulsarConfiguration.getProducer(pulsarClient, pulsarProperties, true,
                "public", MqConstant.NAMESPACE_WXGZH, MqConstant.ACQUISITION_HISTORY_LIST_TOPIC);
        return new PulsarMessageSender(producer);
    }

    @Bean(name = "wxgzhHistoryListDealSender")
    public MqMessageSender getWXGZHHistoryListDealSender() throws PulsarClientException {
        Producer<String> producer = PulsarConfiguration.getProducer(pulsarClient, pulsarProperties, true,
                "public", MqConstant.NAMESPACE_WXGZH, MqConstant.DATA_DEAL_HISTORY_LIST_TOPIC);
        return new PulsarMessageSender(producer);
    }

    @Bean(name = "wxgzhContentAcqSender")
    public MqMessageSender getWXGZHContentAcqSender() throws PulsarClientException {
        Producer<String> producer = PulsarConfiguration.getProducer(pulsarClient, pulsarProperties, true,
                "public", MqConstant.NAMESPACE_WXGZH, MqConstant.ACQUISITION_CONTENT_TOPIC);
        return new PulsarMessageSender(producer);
    }

    @Bean(name = "wxgzhContentDealSender")
    public MqMessageSender getWXGZHContentDealSender() throws PulsarClientException {
        Producer<String> producer = PulsarConfiguration.getProducer(pulsarClient, pulsarProperties, true,
                "public", MqConstant.NAMESPACE_WXGZH, MqConstant.DATA_DEAL_CONTENT_TOPIC);
        return new PulsarMessageSender(producer);
    }

    @Bean(name = "wbHistoryListAcqSender")
    public MqMessageSender getWBHistoryListAcqSender() throws PulsarClientException {
        Producer<String> producer = PulsarConfiguration.getProducer(pulsarClient, pulsarProperties, true,
                "public", MqConstant.NAMESPACE_WB, MqConstant.ACQUISITION_HISTORY_LIST_TOPIC);
        return new PulsarMessageSender(producer);
    }

    @Bean(name = "wbHistoryListDealSender")
    public MqMessageSender getWBHistoryListDealSender() throws PulsarClientException {
        Producer<String> producer = PulsarConfiguration.getProducer(pulsarClient, pulsarProperties, true,
                "public", MqConstant.NAMESPACE_WB, MqConstant.DATA_DEAL_HISTORY_LIST_TOPIC);
        return new PulsarMessageSender(producer);
    }

    @Bean(name = "wbContentAcqSender")
    public MqMessageSender getWBContentAcqSender() throws PulsarClientException {
        Producer<String> producer = PulsarConfiguration.getProducer(pulsarClient, pulsarProperties, true,
                "public", MqConstant.NAMESPACE_WB, MqConstant.ACQUISITION_CONTENT_TOPIC);
        return new PulsarMessageSender(producer);
    }

    @Bean(name = "wbContentDealSender")
    public MqMessageSender getWBContentDealSender() throws PulsarClientException {
        Producer<String> producer = PulsarConfiguration.getProducer(pulsarClient, pulsarProperties, true,
                "public", MqConstant.NAMESPACE_WB, MqConstant.DATA_DEAL_CONTENT_TOPIC);
        return new PulsarMessageSender(producer);
    }

}
